package com.cmcc.pcap.jpcap.udp;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map.Entry;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import jpcap.packet.UDPPacket;


/**
 * 
 * @author Amish
 * @date 2018年12月24日 下午3:45:26
 *
 */
public class UdpConnectionCache {
	public static Logger logger = LogManager.getLogger(UdpConnectionCache.class);
	public AtomicInteger totalPacketNum;//缓冲区收到的所有UDP Packet数量
	public AtomicInteger pickupPacketNum;//缓冲区已经提取出的UDP Packet数量
	public AtomicInteger handledPacketNum;//被工作线程/消费者线程 “处理完”的UDP Packet数量
	
	LinkedHashMap <String ,UdpConnection> chm;
	
	public UdpConnectionCache(){
		totalPacketNum = new AtomicInteger(0);
		pickupPacketNum = new AtomicInteger(0);
		handledPacketNum = new AtomicInteger(0);
		this.chm = new LinkedHashMap <String ,UdpConnection>();
	}
	
	public synchronized void addUDPPacket(UDPPacket udpPacket){
		this.totalPacketNum.getAndIncrement();
		while( chm.size() >= 5 ){//TODO 如何更合理的选择这个控制条件？
			try {
				this.wait();
			} catch (InterruptedException e) {
				logger.error(e.getCause(),e);
			}
		}
		
		try{
			
			
			//每收到一个UDP包，生成4元组对应的一个Key（注意消息方向），唯一标识一个UDP连接
			String src_ip = udpPacket.src_ip.getHostAddress();
			String src_port = Integer.toString(udpPacket.src_port);
			String dst_ip = udpPacket.dst_ip.getHostAddress();
			String dst_port =  Integer.toString(udpPacket.dst_port);
			String src = src_ip+":"+src_port;
			String dst = dst_ip+":"+dst_port;
			String connectionKey = src+"-"+dst;       //通过四元组生成connectionKey
			String connectionKeyReverse = dst+"-"+src;//通过四元组生成connectionKey，源目的对调
			
			UdpConnection udpConnection = chm.get(connectionKey) !=null? chm.get(connectionKey) : chm.get(connectionKeyReverse);
			if( udpConnection == null ){
				logger.debug(connectionKey);
				udpConnection = new UdpConnection();
				udpConnection.connectionKey = connectionKey;
				udpConnection.lastPacketTime = System.currentTimeMillis();
				udpConnection.al.add(udpPacket);
				
				chm.put(connectionKey, udpConnection);
			}else{
				udpConnection.lastPacketTime = System.currentTimeMillis();
				udpConnection.al.add(udpPacket);
			}

			UdpProtocol udpProtocol = UdpProtocolFilter.filterUdpProtocol(udpPacket) ;
			udpConnection.protocol = udpProtocol;
			
		}catch(Exception e){
			logger.error(e.getCause(), e);
		}
		this.notify();

	}
	
	public synchronized ArrayList<UdpConnection> getUdpConnection(){
		ArrayList<UdpConnection> al = new ArrayList<UdpConnection>();
		try{
			while(chm.values().size() == 0){//TODO 如何选择这个控制条件？
				try {
					this.wait();
				} catch (InterruptedException e) {
					logger.error(e.getCause(),e);
				}
			}
			
			int i = 2 ;
			Iterator<Entry<String,UdpConnection>> iterator = chm.entrySet().iterator();
			while(iterator.hasNext()) {
				if(i==0) {
					this.notify();
					return al;
				}
				
				UdpConnection udpConnection = iterator.next().getValue();
				if(udpConnection.al.size() >=8 || ((System.currentTimeMillis() - udpConnection.lastPacketTime) > 5000) ){
					al.add(udpConnection);
					iterator.remove();
					this.pickupPacketNum.getAndAdd(udpConnection.al.size());
					i--;
				}
			}
			
		}catch(Exception e){
			logger.error(e.getCause(),e);
			//receiveInvalidTcpPacketNum.getAndIncrement();
		}
		this.notify();
		return al;
	}

}
